package com.wfg.flink.core.example;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class WatermarkDemo {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 添加数据源，这里以简单的整数流为例
        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // 定义 Watermark 策略，允许 2 秒的乱序
        WatermarkStrategy<Integer> watermarkStrategy = WatermarkStrategy
                .<Integer>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((SerializableTimestampAssigner<Integer>) (element, l) -> element);

        // 应用 Watermark 策略，并处理数据流
        DataStream<Integer> watermarkedStream = dataStream
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .keyBy(value -> value)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 使用 5 秒的滚动窗口进行聚合操作
                .sum("sum"); // 对窗口内的元素进行求和操作

        // 输出结果
        watermarkedStream.print();

        // 执行任务
        env.execute("Watermark Demo");
    }
}